package com.china.produce.config.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.protocol.FlatMessage;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @ClassName
 * @Author niujie
 * @Version
 * @Description kafka监听器canal数据
 * @CreateTime 2025/6/8
 */
@Component
public class CanalKafkaListener {

    @KafkaListener(topics = "test_example", groupId = "canal-group")
    public void handleMessage(String message, Acknowledgment ack) {
        FlatMessage flatMessage = JSON.parseObject(message, FlatMessage.class);
        System.out.println("接收到canal flatMessage数据：" + flatMessage);

        // 示例：获取变更数据
        if (flatMessage.getData() != null) {
            flatMessage.getData().forEach(row -> {
                System.out.println("行数据：" + row);
            });
        }
        ack.acknowledge();
    }
}
